package events

import (
	"context"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/saramax"
	"jk-time/webook/search/domain"
	"jk-time/webook/search/service"
	"time"
)

type UserEvent struct {
	Id       int64  `json:"id"`
	Email    string `json:"email"`
	Nickname string `json:"nickname"`
	Phone    string `json:"phone"`
}
type UserKafkaComsumer struct {
	client  sarama.Client
	l       logger.Logger
	syncSvc service.SyncService
}

func (UserEvent) Topic() string {
	return "search_Users"
}

func NewUserKafkaComsumer(client sarama.Client, l logger.Logger, syncSvc service.SyncService) Consumer {
	return &UserKafkaComsumer{
		client:  client,
		syncSvc: syncSvc,
		l:       l,
	}
}
func (k *UserKafkaComsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("search_User_1", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{UserEvent{}.Topic()},
			saramax.NewHandler[UserEvent]("search_User_1", k.Consume))
		if err != nil {
			zap.L().Error("退出了消费循环,异常", zap.Error(err))
		}
	}()
	return err
}

func (k *UserKafkaComsumer) Consume(msg *sarama.ConsumerMessage, t UserEvent) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	zap.L().Info("kafka:", zap.String("topic", t.Topic()),
		zap.String("value", string(msg.Value)))
	return k.syncSvc.InputUser(ctx, domain.User{
		Id:       t.Id,
		Email:    t.Email,
		Nickname: t.Nickname,
		Phone:    t.Phone,
	})
}
